#include "thread_pool.h"

#include "util.h"

ThreadPool::ThreadPool(const int threadNums)
{
    for (int i = 0; i < threadNums; i++)
    {
        threads.emplace_back(&ThreadPool::threadFun, this, i);
    }
}

ThreadPool::~ThreadPool()
{
    logDebug("~ThreadPool()");
    stop.store(true, std::memory_order_relaxed);

    {
        std::unique_lock<std::mutex> lock(mtx);
        tasks = std::queue<std::function<void()>>();
    }

    cvTask.notify_all();

    for (int i = 0; i < threads.size(); i++)
    {
        threads[i].join();
    }
    logDebug("~ThreadPool() finish");
}

void ThreadPool::waitAll()
{
    std::unique_lock<std::mutex> lock(mtx);

    cvFinish.wait(lock, [this]()
                  { return tasks.empty() && pendingTasks.load(std::memory_order_relaxed) == 0; });
}

void ThreadPool::addTask(std::function<void()>&& task)
{
    {
        std::unique_lock<std::mutex> lock(mtx);
        tasks.emplace(std::move(task));
    }
    cvTask.notify_all();
}

void ThreadPool::threadFun(const int id)
{
    logDebug("threadFun %d starting", id);

    while (true)
    {
        std::function<void()> task;
        {
            std::unique_lock<std::mutex> lock(mtx);
            cvTask.wait(
                lock, [this]() { return stop.load(std::memory_order_relaxed) || !tasks.empty(); });

            // 析构函数才会设置stop
            if (stop.load(std::memory_order_relaxed))
            {
                logDebug("threadFun %d: stop == true", id);
                return;
            }

            task = std::move(tasks.front());
            tasks.pop();
        }

        ++pendingTasks;
        task();
        --pendingTasks;

        {
            std::unique_lock<std::mutex> lock(mtx);
            if (tasks.empty() && pendingTasks.load(std::memory_order_relaxed) == 0)
            {
                cvFinish.notify_one();
            }
        }
    }

    logDebug("threadFun %d exit", id);
}
